package com.kafka.consumer;



import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.example.error.NotRetryableException;
import org.example.error.RetryableException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.support.serializer.JsonSerializer;
import org.springframework.util.backoff.FixedBackOff;
import org.springframework.web.client.HttpServerErrorException;

import java.util.HashMap;
import java.util.Map;

/**
 * @author Administrator
 */
@Configuration
public class KafkaConsumerConfig {
    @Autowired
    private Environment environment;

    //    @Bean
//    public ConsumerFactory<String, Object> consumerFactory() {
//        Map<String, Object> config = new HashMap<>();
//        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, environment.getProperty("spring.kafka.consumer.bootstrap-servers"));
//        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//        //ErrorHandlingDeserializer 是JsonDeserializer的装饰器实现，用于处理消费者消息序列化错误
//        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
//        //实际的反序列化应用用JsonDeserializer完成  通过该配置 允许消费者序列化错误后，继续执行并消费来自主题的下一条消息
//        config.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);
//        config.put(JsonDeserializer.TRUSTED_PACKAGES, environment.getProperty("spring.kafka.consumer.properties.spring.json.trusted.packages"));
//        config.put(ConsumerConfig.GROUP_ID_CONFIG, environment.getProperty("spring.kafka.consumer.group-id"));
//        return new DefaultKafkaConsumerFactory<>(config);
//    }
    @Bean(name = "consumerFactoryA")
    ConsumerFactory<String, Object> consumerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                environment.getProperty("spring.kafka.consumer.bootstrap-servers"));
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
        config.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);
        config.put(JsonDeserializer.TRUSTED_PACKAGES,
                environment.getProperty("spring.kafka.consumer.properties.spring.json.trusted.packages"));
        config.put(ConsumerConfig.GROUP_ID_CONFIG, environment.getProperty("spring.kafka.consumer.group-id"));
        // 设置分区分配策略为 StickyAssignor
        config.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.StickyAssignor");
        config.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, environment.getProperty("spring.kafka.consumer.properties.max.poll.interval.ms"));

        // 设置从最早的偏移量开始读取
       // config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
      //  config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 设置从最新的偏移量开始读取
       // props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        // 设置 poll 相关的属性
       // config.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 5000); // 最大轮询间隔时间（毫秒）
       // config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);      // 每次轮询的最大记录数
        return new DefaultKafkaConsumerFactory<>(config);
    }

    //    @Bean
//    ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory(
//            ConsumerFactory<String, Object> consumerFactory) {
//
//        ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
//        factory.setConsumerFactory(consumerFactory);
//
//        return factory;
//    }
    @Bean
    ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory(
            @Qualifier(value = "consumerFactoryA") ConsumerFactory<String, Object> consumerFactory, @Qualifier(value = "deadkafkaTemplate") KafkaTemplate<String, Object> kafkaTemplate) {

        DefaultErrorHandler errorHandler = new DefaultErrorHandler(new DeadLetterPublishingRecoverer(kafkaTemplate),
                //实际间隔 最大重试次数
                new FixedBackOff(5000,3));
        //添加的异常不会被重试
        errorHandler.addNotRetryableExceptions(NotRetryableException.class, HttpServerErrorException.class);
        //注册重试异常
        errorHandler.addRetryableExceptions(RetryableException.class);
        ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setCommonErrorHandler(errorHandler);
       // factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE); // 设置为手动确认
      //  factory.setBatchListener(true); // 启用批量监听

        return factory;
    }

    @Bean(name = "deadkafkaTemplate")
    KafkaTemplate<String, Object> deadkafkaTemplate(@Qualifier(value = "deadproducerFactory") ProducerFactory<String, Object> producerFactory) {
        return new KafkaTemplate<>(producerFactory);
    }

    @Bean(name = "deadproducerFactory")
    ProducerFactory<String, Object> deadproducerFactory() {
        Map<String, Object> config = new HashMap<>(5);
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, environment.getProperty("spring.kafka.consumer.bootstrap-servers"));
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        return new DefaultKafkaProducerFactory<>(config);
    }
}
